package miaodongsql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 根据数据源更新热词榜单
 */
public class D2_hotword {

  public static void main(String[] args) {

    Configuration configuration = new Configuration();
    configuration.setString("rest.port", "9091");
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
    env.setParallelism(1);
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


    env.enableCheckpointing(10 * 1000);
    env.setStateBackend(new EmbeddedRocksDBStateBackend());

    CheckpointConfig checkpointConfig = env.getCheckpointConfig();
    checkpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
    checkpointConfig.setCheckpointStorage("file:///flink_ckp");


    //
    // String genSql = "CREATE TABLE ods_tb ( " +
    //     " stime STRING," +
    //     " word STRING," +
    //     " rowtime AS to_timestamp(stime)," +
    //     " WATERMARK FOR rowtime AS rowtime - interval '30' second" +
    //     ") WITH ( " +
    //     "  'connector' = 'kafka'," +
    //     "  'topic' = 'test'," +
    //     "  'properties.bootstrap.servers' = 'wsl:9092'," +
    //     "  'properties.group.id' = 'testGroup'," +
    //     "  'scan.startup.mode' = 'latest-offset'," +
    //     "  'format' = 'json'" +
    //     ")";

    String genSql = "CREATE TABLE ods_tb ( " +
        " stime STRING," +
        " word STRING" +
        ") WITH ( " +
        " 'connector'='datagen', " +
        " 'rows-per-second'='10000'," +
        " 'fields.stime.kind'='random'," +
        " 'fields.word.kind'='random'," +
        " 'fields.word.length'='5'" +
        ")";


    String print_console = "CREATE TABLE print (" +
        "    snkey STRING, " +
        "    index BIGINT, " +
        "    word STRING" +
        ") WITH (" +
        "    'connector' = 'print'" +
        ")";


    String print_redis = "CREATE TABLE redis (" +
        "    snkey STRING, " +
        "    index BIGINT, " +
        "    word STRING, " +
        " primary key (snkey) not ENFORCED" +
        ") WITH (" +
        "    'connector' = 'redis',\n" +
        "    'host' = 'bdyun-suzhou-tw04f0163',\n" +
        "    'port' = '6379',\n" +
        "    'redis-mode' = 'single',\n" +
        "    'timeout' = '2000',\n" +
        "    'password' = 'yy2024',\n" +
        "    'command' = 'HMSET',\n" +
        "    'max.retries' = '5',\n" +
        "    'database' = '5',\n" +
        "    'ttl' = '10800'" +
        ")";



    String sql = " CREATE VIEW v_base AS " +
        "SELECT word,count(1) cnt from ods_tb GROUP BY word";

    String sql1 = "INSERT INTO redis " +
        " SELECT 'HOT_WORD' as snkey, " +
        " rownum, " +
        " word " +
        " from " +
        " (SELECT *,row_number()OVER(PARTITION BY 'a' ORDER BY cnt desc) as rownum FROM v_base) where rownum < 11";


    tableEnv.executeSql(genSql);
    tableEnv.executeSql(print_redis);
    tableEnv.executeSql(sql);
    tableEnv.executeSql(sql1);


  }
}
